package com.csap.framework.integration.listener;

import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.shade.org.apache.commons.lang3.StringUtils;
import com.csap.framework.integration.MessageFilterType;
import com.csap.framework.integration.RocketmqConfig;
import com.csap.framework.integration.consumer.EnableRocketmqBean;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.ApplicationContext;

import javax.annotation.Resource;

/**
 * 默认消息监听
 *
 * @author ycf
 */

@Slf4j
public class SimpleMessageListener implements com.aliyun.openservices.ons.api.MessageListener {
    @Resource
    private ApplicationContext applicationContext;
    @Resource
    private EnableRocketmqBean enableRocketmqBean;
    @Resource
    private RocketmqConfig rocketmqConfig;

    @Override
    public Action consume(Message message, ConsumeContext context) {
        log.debug("messageFilterType is {} ,product:{} consume :{}", rocketmqConfig.getConsumer().getProducerMessageFilterType(), rocketmqConfig.getConsumer().getConsumerMessageFilterType(), message.toString());
        String string;
        if (rocketmqConfig.getConsumer().getProducerMessageFilterType().equals(MessageFilterType.TOPIC)) {
            string = message.getTopic();
        } else if (rocketmqConfig.getConsumer().getProducerMessageFilterType().equals(MessageFilterType.TAG)) {
            string = message.getTag();
        } else {
            string = message.getKey();
        }
        log.debug("filter value :{},enableRocketmqBean map keys :{}", string, enableRocketmqBean.getRocketmqBeans());
        EnableRocketmqBean.RocketmqBean rocketmqBean = enableRocketmqBean.filter(string, rocketmqConfig.getConsumer().getConsumerMessageFilterType());
        if (StringUtils.isEmpty(string) || rocketmqBean == null) {
            return Action.ReconsumeLater;
        }
        return applicationContext.getBean(rocketmqBean.getAClass()).consume(message, context);
    }
}
